package com.we.risk.phoneBook.adm;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.we.flink.utils.WeKafkaPropertyReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.util.Iterator;
import java.util.Optional;
import java.util.Properties;

public class AdmUserLabelCmpTest {
    public static final String RELEASEPROP = "risk/phonebook/adm_user_label_kfk_cmp.properties";
    public static final String PROPSETS = "risk/phonebook/phonebook230126.properties";
    public static Logger LOG = LoggerFactory.getLogger(AdmUserLabelCmpTest.class);

    public static final String DATEFORMAT = "yyyy-MM-dd";
    public static final String INPUTKEYBY = "user_key";
    public static final String TIMESTAMP = "timestamp";
    public static final String CAPTURETIME = "capture_time";
    public static final String OUTKEYBY = "user_key";
    public static final String LABELOUTIME = "label_output_time";
    public static final String OUTJOBID = "job_id";
    public static final String INPUTJOBID = "jobID";
    public static final String PHONEBOOKLIST = "phoneBookList";

    public static void main(String[] args) throws IOException {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        try {
            WeKafkaPropertyReader paramReader = WeKafkaPropertyReader.init(RELEASEPROP);
            /** RocksDB */
            env.setStateBackend(new RocksDBStateBackend(paramReader.getRocksDBBackendUrl()));
            /** checkpoint configure */
            CheckpointConfig ckConf = env.getCheckpointConfig();
            ckConf.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            ckConf.setCheckpointInterval(10 * 60 * 1000); // ms
            ckConf.setCheckpointTimeout(60 * 60 * 1000);
            ckConf.setMaxConcurrentCheckpoints(1);
            ckConf.setMinPauseBetweenCheckpoints(500);
            ckConf.enableExternalizedCheckpoints(
                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

            ckConf.enableUnalignedCheckpoints();

            /** Consumer Kafka */
            String srcKafkaTopic = paramReader.getTmpKfkTopic();
            String srcKafkaBootStrapServer = paramReader.getTmpKfkBootStrapServer();
            String srckfkGrupId = paramReader.getTmpKfkGroupId();

            Properties consumProp = new Properties();
            consumProp.setProperty("group.id", srckfkGrupId);
            consumProp.setProperty(
                    "key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            consumProp.setProperty(
                    "value.deserializer",
                    "org.apache.kafka.common.serialization.StringDeserializer");

            KafkaSource<String> kfkSource =
                    KafkaSource.<String>builder()
                            .setBootstrapServers(srcKafkaBootStrapServer)
                            .setTopics(srcKafkaTopic)
                            .setStartingOffsets(OffsetsInitializer.earliest())
                            .setProperties(consumProp)
                            .setValueOnlyDeserializer(new SimpleStringSchema())
                            .build();

            SingleOutputStreamOperator<String> input =
                    env.fromSource(kfkSource, WatermarkStrategy.noWatermarks(), "PhoneRecord")
                            .uid("phone-book-source");

            SingleOutputStreamOperator<String> out =
                    input.filter(
                                    new FilterFunction<String>() {
                                        @Override
                                        public boolean filter(String value) throws Exception {
                                            JSONObject inputJson = JSONObject.parseObject(value);
                                            if (inputJson.containsKey(INPUTKEYBY)
                                                    && inputJson.getString(INPUTKEYBY) != null) {
                                                return true;
                                            }
                                            return false;
                                        }
                                    })
                            .keyBy(
                                    new KeySelector<String, String>() {
                                        @Override
                                        public String getKey(String value) throws Exception {
                                            JSONObject inputJson = JSONObject.parseObject(value);
                                            return inputJson.getString(INPUTKEYBY);
                                        }
                                    })
                            .flatMap(new UserLabelRichFlatMapFunc())
                            .keyBy(
                                    new KeySelector<String, String>() {
                                        @Override
                                        public String getKey(String value) throws Exception {
                                            JSONObject inputJson = JSONObject.parseObject(value);
                                            return inputJson.getString(OUTKEYBY);
                                        }
                                    })
                            .flatMap(
                                    new RichFlatMapFunction<String, String>() {
                                        ValueState<JSONObject> prStage2UserLabel;

                                        @Override
                                        public void open(Configuration parameters)
                                                throws Exception {
                                            prStage2UserLabel =
                                                    getRuntimeContext()
                                                            .getState(
                                                                    new ValueStateDescriptor<
                                                                            JSONObject>(
                                                                            "PRStage2UserLabel",
                                                                            JSONObject.class));
                                        }

                                        @Override
                                        public void flatMap(String value, Collector<String> out)
                                                throws Exception {
                                            try {
                                                JSONObject inputJson =
                                                        JSONObject.parseObject(value);
                                                JSONObject tmpState = prStage2UserLabel.value();
                                                JSONObject curState =
                                                        tmpState == null
                                                                ? initPropJson(PROPSETS)
                                                                : tmpState;

                                                String outKey = inputJson.getString(OUTKEYBY);
                                                Long input_cap_time =
                                                        inputJson.getLong(CAPTURETIME);
                                                Long state_cap_time = curState.getLong(CAPTURETIME);

                                                if (input_cap_time > state_cap_time) {
                                                    Iterator<String> iterator =
                                                            curState.keySet().iterator();
                                                    while (iterator.hasNext()) {
                                                        String key = iterator.next();
                                                        if (!key.equals(OUTKEYBY)
                                                                && !key.equals(CAPTURETIME)) {
                                                            Object labelV = inputJson.get(key);
                                                            Object state_v = curState.get(key);
                                                            if (state_v != null && labelV != null) {
                                                                if (!state_v.equals(labelV)) {
                                                                    JSONObject outJson =
                                                                            new JSONObject();
                                                                    outJson.put(OUTKEYBY, outKey);
                                                                    outJson.put(
                                                                            CAPTURETIME,
                                                                            input_cap_time);
                                                                    outJson.put("label_name", key);
                                                                    outJson.put(
                                                                            "label_value", labelV);
                                                                    out.collect(outJson.toString());
                                                                    curState.put(key, labelV);
                                                                }
                                                            } else {
                                                                if (state_v != labelV) {
                                                                    JSONObject outJson =
                                                                            new JSONObject();
                                                                    outJson.put(OUTKEYBY, outKey);
                                                                    outJson.put(
                                                                            CAPTURETIME,
                                                                            input_cap_time);
                                                                    outJson.put("label_name", key);
                                                                    outJson.put(
                                                                            "label_value", labelV);
                                                                    out.collect(
                                                                            outJson.toString(
                                                                                    SerializerFeature
                                                                                            .WriteMapNullValue));
                                                                    curState.put(key, labelV);
                                                                }
                                                            }
                                                        }
                                                    }
                                                    curState.put(CAPTURETIME, input_cap_time);
                                                    prStage2UserLabel.update(curState);
                                                }
                                            } catch (Exception e) {
                                                LOG.error(
                                                        "RichFlatMapFunction Exception: "
                                                                + e.toString());
                                                LOG.error(
                                                        "RichFlatMapFunction excep input: "
                                                                + value);
                                                //
                                                // LOG.error("RichFlatMapFunction Exception input: "
                                                // + (value.length() >= 100 ? value.substring(0,
                                                // 100) : value));
                                                e.printStackTrace();
                                            }
                                        }
                                    });

            /** sink to Kafka */
            String sinkkfkTopic = paramReader.getKfkTopic();
            Properties sinkProp = new Properties();
            sinkProp.setProperty("bootstrap.servers", paramReader.getKfkBootStrapServer());
            int sinkkfkPartitions = paramReader.getKfkPartitions();

            FlinkKafkaProducer<String> kafkaProducer =
                    new FlinkKafkaProducer<String>(
                            sinkkfkTopic,
                            new WeKafkaKeyedSerializationSchema(),
                            sinkProp,
                            Optional.of(new WeKafkaCustomPartitioner()));

            out.addSink(kafkaProducer).setParallelism(sinkkfkPartitions);

            env.execute(AdmUserLabelCmpTest.class.toString());
        } catch (Exception e) {
            System.out.println("Exception: " + e.getMessage());
            LOG.error("Exception: " + e.getMessage());
            e.printStackTrace();
        }
    }

    private static class UserLabelRichFlatMapFunc extends RichFlatMapFunction<String, String> {
        ValueState<JSONObject> dimUserVState;

        @Override
        public void open(Configuration parameters) throws Exception {
            dimUserVState =
                    getRuntimeContext()
                            .getState(
                                    new ValueStateDescriptor<JSONObject>(
                                            "dimUserVState", JSONObject.class));
        }

        @Override
        public void flatMap(String value, Collector<String> out) throws Exception {
            try {
                JSONObject tmpState = dimUserVState.value();
                JSONObject curState = (tmpState == null ? new JSONObject() : tmpState);

                if (value.contains(INPUTJOBID)) {
                    /** phoneBooklist */
                    LOG.debug(
                            "Debug curState.size: "
                                    + curState.size()
                                    + " curstate: "
                                    + curState.toString());
                    if (curState.size() > 1) {
                        JSONObject res = ProcessData(value, curState);
                        out.collect(res.toString());
                    }
                } else {
                    LOG.debug("Debug: dim UserKey come in: " + value);
                    JSONObject inputJson = JSONObject.parseObject(value);
                    if (!curState.equals(inputJson)) {
                        dimUserVState.update(inputJson);
                    }
                }
            } catch (Exception e) {
                LOG.error("UserLabelRichFlatMapFunc Exception: " + e.toString());
                LOG.error(
                        "UserLabelRichFlatMapFunc Exception input: "
                                + (value.length() >= 100 ? value.substring(0, 100) : value));
                e.printStackTrace();
            }
        }
    }

    private static class WeKafkaKeyedSerializationSchema
            implements KeyedSerializationSchema<String> {
        @Override
        public byte[] serializeKey(String element) {
            JSONObject jsonObject = JSONObject.parseObject(element);
            String keyby = jsonObject.getString(OUTKEYBY);
            return keyby.getBytes();
        }

        @Override
        public byte[] serializeValue(String element) {
            return element.getBytes();
        }

        @Override
        public String getTargetTopic(String element) {
            return null;
        }
    }

    private static class WeKafkaCustomPartitioner extends FlinkKafkaPartitioner<String> {

        @Override
        public int partition(
                String record, byte[] key, byte[] value, String targetTopic, int[] partitions) {
            int partition = Math.abs(new String(key).hashCode() % partitions.length);
            if (LOG.isDebugEnabled()) {
                LOG.info(
                        " partitions: "
                                + partitions.length
                                + " partition: "
                                + partition
                                + " key: "
                                + new String(key));
            }
            return partition;
        }
    }

    public static JSONObject ProcessData(String value, JSONObject userState) {

        JSONObject inputJson = JSONObject.parseObject(value);
        Long captureTime = inputJson.getLong(TIMESTAMP);
        String user_key = inputJson.getString(INPUTKEYBY);
        String job_id = inputJson.getString(INPUTJOBID);
        JSONArray phoneBookList = inputJson.getJSONArray(PHONEBOOKLIST);
        int adb_cnt = phoneBookList.size();
        LOG.debug("Debug adb_cnt: " + adb_cnt);

        userBasicInfo basicInfo = new userBasicInfo();
        for (int i = 0; i < adb_cnt; i++) {
            phoneBookObject phoneBookObject =
                    new phoneBookObject((JSONObject) phoneBookList.get(i), userState);
            basicInfo.process(phoneBookObject);
        }

        JSONObject resJson = new JSONObject();
        resJson.put(OUTKEYBY, user_key);
        resJson.put(CAPTURETIME, captureTime);
        resJson.put(OUTJOBID, job_id);
        basicInfo.format(resJson);
        long label_output_time = System.currentTimeMillis();
        resJson.put(LABELOUTIME, label_output_time);
        //            LOG.warn("Debug resJson size: " + resJson.size() );
        //            LOG.warn("Debug resJson: " + resJson.toString());
        return resJson;
    }

    private static JSONObject initPropJson(String propFile) throws IOException {
        InputStream resourceAsStream =
                AdmUserLabelCmpTest.class.getClassLoader().getResourceAsStream(propFile);
        ParameterTool reader = ParameterTool.fromPropertiesFile(resourceAsStream);
        JSONObject res = new JSONObject();
        Iterator<Object> keysIterator = reader.getProperties().keySet().iterator();
        while (keysIterator.hasNext()) {
            res.put((String) keysIterator.next(), -1);
        }
        return res;
    }

    public static void main1(String[] args) throws IOException {
        String value =
                "{\"adb_tel_cnt\":0,\"adb_good1_rte\":0.0,\"adb_tel_spc_cnt\":0,\"adb_good1_cnt\":0,\"adb_phn_cnt\":0,\"user_key\":\"bcee845df29b59d6adfcc65190fb2c51\",\"adb_tel_spc_rte\":0.0,\"adb_lon_cnt\":0,\"adb_zh_num_cnt\":1,\"adb_zh_num_rte\":1.0,\"adb_zh_num_max\":2,\"adb_lon_rte\":0.0,\"adb_zh_num_min\":2,\"adb_zh_num_avg\":2.0,\"capture_time\":1674622535971,\"adb_nam_cnt\":1,\"adb_nam_rte\":1.0,\"adb_num_num_min\":0,\"adb_num_num_avg\":0.0,\"adb_en_num_min\":0,\"adb_en_num_avg\":0.0,\"adb_nam_len_max\":2,\"adb_good2_rte\":0.0,\"adb_tel_9596_rte\":0.0,\"adb_en_num_cnt\":1,\"adb_num_num_cnt\":1,\"adb_tel_400_rte\":0.0,\"adb_tel_9596_cnt\":0,\"adb_num_num_rte\":1.0,\"job_id\":\"haohuan-5031c92f-701d-eace-c951-9dfa61799d32\",\"adb_num_num_max\":0,\"adb_fmy_cnt\":1,\"adb_good2_cnt\":0,\"adb_nam_len_min\":2,\"adb_nam_len_avg\":2.0,\"adb_bad_rte\":0.0,\"adb_tel_rte\":0.0,\"adb_cnt\":1,\"adb_tel_400_cnt\":0,\"adb_en_num_max\":0,\"adb_fmy_rte\":1.0,\"adb_phn_rte\":0.0,\"adb_en_num_rte\":1.0,\"label_output_time\":1674964739252,\"adb_bad_cnt\":0}";
        try {
            JSONObject inputJson = JSONObject.parseObject(value);
            JSONObject curState = initPropJson(PROPSETS);

            String outKey = inputJson.getString(OUTKEYBY);
            Long input_cap_time = inputJson.getLong(CAPTURETIME);
            Long state_cap_time = curState.getLong(CAPTURETIME);

            if (input_cap_time > state_cap_time) {
                //                            Iterator<String> iterator =
                // inputJson.keySet().iterator();
                Iterator<String> iterator = curState.keySet().iterator();
                while (iterator.hasNext()) {
                    String key = iterator.next();
                    if (!key.equals(OUTKEYBY) && !key.equals(CAPTURETIME)) {
                        Object labelV = inputJson.get(key);
                        Object state_v = curState.get(key);
                        if (state_v != null && labelV != null) {
                            if (!state_v.equals(labelV)) {
                                JSONObject outJson = new JSONObject();
                                outJson.put(OUTKEYBY, outKey);
                                outJson.put(CAPTURETIME, input_cap_time);
                                outJson.put("label_name", key);
                                outJson.put("label_value", labelV);
                                System.out.println(outJson.toString());
                                curState.put(key, labelV);
                            }
                        } else {
                            if (state_v != labelV) {
                                JSONObject outJson = new JSONObject();
                                outJson.put(OUTKEYBY, outKey);
                                outJson.put(CAPTURETIME, input_cap_time);
                                outJson.put("label_name", key);
                                outJson.put("label_value", labelV);
                                System.out.println(
                                        outJson.toString(SerializerFeature.WriteMapNullValue));
                                curState.put(key, labelV);
                            }
                        }
                    }
                }
                curState.put(CAPTURETIME, input_cap_time);
            }
        } catch (Exception e) {
            LOG.error("RichFlatMapFunction Exception: " + e.toString());
            LOG.error("RichFlatMapFunction excep input: " + value);
            //                        LOG.error("RichFlatMapFunction Exception input: " +
            // (value.length() >= 100 ? value.substring(0, 100) : value));
            e.printStackTrace();
        }
    }
}
